package com.gomro.galaxy.foundation.bi.engine.analysis;

import cn.gomro.cloud.core.common.utils.DateUtils;
import cn.gomro.cloud.core.common.utils.EncryptUtils;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.gomro.galaxy.core.batch.ElasticSearchClzItemReader;
import com.gomro.galaxy.core.batch.ElasticSearchClzItemWriter;
import com.gomro.galaxy.core.commons.elastic.ElasticSearchRest;
import com.gomro.galaxy.foundation.bi.domain.Constants;
import com.gomro.galaxy.foundation.bi.domain.event.VendorTradingVolumeEvent;
import com.gomro.galaxy.foundation.bi.domain.model.*;
import com.gomro.galaxy.foundation.bi.domain.model.analysis.VendorTradingVolume;
import com.gomro.galaxy.foundation.bi.engine.AnalysisLastTime;
import com.gomro.galaxy.foundation.bi.engine.model.AnalysisCache;
import com.gomro.galaxy.foundation.bi.engine.query.OrderGoodsRefundQuery;
import com.gomro.galaxy.foundation.bi.engine.query.OrganizationQuery;
import com.gomro.galaxy.foundation.bi.engine.query.QuotationGoodsQuery;
import com.gomro.galaxy.foundation.bi.engine.query.VendorOrderQuery;
import com.gomro.galaxy.foundation.bi.engine.util.BoolQueryBuilders;
import com.gomro.galaxy.foundation.bi.engine.util.EngineUtils;
import com.gomro.galaxy.foundation.bi.exception.BiException;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.aggregations.Aggregation;
import org.elasticsearch.search.aggregations.AggregationBuilder;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.metrics.Sum;
import org.elasticsearch.search.aggregations.metrics.SumAggregationBuilder;
import org.elasticsearch.search.aggregations.metrics.ValueCount;
import org.elasticsearch.search.aggregations.metrics.ValueCountAggregationBuilder;
import org.elasticsearch.search.sort.SortBuilders;
import org.elasticsearch.search.sort.SortOrder;
import org.springframework.batch.core.*;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepScope;
import org.springframework.batch.core.launch.support.RunIdIncrementer;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.cloud.bus.BusProperties;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.lang.NonNull;
import org.springframework.util.CollectionUtils;

import java.lang.reflect.Array;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.*;

import static com.gomro.galaxy.foundation.bi.engine.util.EngineUtils.MAX_BATCH_NUM;

/**
 * 供应商交易量统计job
 */
@Slf4j
@Configuration
public class VendorTradingJob {
    private static final String BiAnalysisLastName = Constants.BiDataMartIndexPrefix + AnalysisLastTime.TableName;
    private static final String BiWarehouseOrganizationName = Constants.BiWarehouseIndexPrefix + AcquireOrganization.TableName;
    private static final String BiWarehouseQuotationGoodsName = Constants.BiWarehouseIndexPrefix + AcquireQuotationGoods.TableName;
    private static final String BiWarehouseOrderGoodsName = Constants.BiWarehouseIndexPrefix + AcquireOrderGoods.TableName;
    private static final String BiWarehouseVendorOrderName = Constants.BiWarehouseIndexPrefix + AcquireVendorOrder.TableName;
    private static final String BiWarehouseOrderGoodsRefundName = Constants.BiWarehouseIndexPrefix + AcquireOrderGoodsRefund.TableName;
    private static final String BiDataMartName = Constants.BiDataMartIndexPrefix + VendorTradingVolume.TableName;
    private static final String JobName = Constants.JobBeanNamePrefix + VendorTradingVolume.TableName;
    private static final String StepName1 = Constants.StepBeanNamePrefix + VendorTradingVolume.TableName + 1;
    private static final String StepName2 = Constants.StepBeanNamePrefix + VendorTradingVolume.TableName + 2;
    private static final String StepName3 = Constants.StepBeanNamePrefix + VendorTradingVolume.TableName + 3;
    private static final String StepName4 = Constants.StepBeanNamePrefix + VendorTradingVolume.TableName + 4;

    private static final String ReaderName1 = Constants.ReaderBeanNamePrefix + VendorTradingVolume.TableName + 1;
    private static final String ReaderName2 = Constants.ReaderBeanNamePrefix + VendorTradingVolume.TableName + 2;

    private static final String WriterName1 = Constants.WriterBeanNamePrefix + VendorTradingVolume.TableName + 1;
    private static final String WriterName2 = Constants.WriterBeanNamePrefix + VendorTradingVolume.TableName + 2;

    private static final String ProcessName1 = Constants.ProcessBeanNamePrefix + VendorTradingVolume.TableName + 1;
    private static final String ProcessName2 = Constants.ProcessBeanNamePrefix + VendorTradingVolume.TableName + 2;
    private static final String ProcessName3 = Constants.ProcessBeanNamePrefix + VendorTradingVolume.TableName + 3;
    private static final String ProcessName4 = Constants.ProcessBeanNamePrefix + VendorTradingVolume.TableName + 4;

    private ApplicationEventPublisher applicationEventPublisher;
    private BusProperties busProperties;
    private JobBuilderFactory jobBuilderFactory;
    private StepBuilderFactory stepBuilderFactory;
    private ElasticSearchRest rest;
    private RestHighLevelClient client;
    private ObjectMapper mapper;
    private AnalysisCache<VendorTradingVolume> cache = new AnalysisCache<>();

    public VendorTradingJob(ApplicationEventPublisher applicationEventPublisher, BusProperties busProperties, JobBuilderFactory jobBuilderFactory, StepBuilderFactory stepBuilderFactory, RestHighLevelClient client, ObjectMapper mapper) {
        this.applicationEventPublisher = applicationEventPublisher;
        this.busProperties = busProperties;
        this.jobBuilderFactory = jobBuilderFactory;
        this.stepBuilderFactory = stepBuilderFactory;
        this.client = client;
        this.mapper = mapper;
        this.rest = new ElasticSearchRest(client, mapper);
    }

    private JobExecutionListener jobExecutionListener = new JobExecutionListener() {

        @Override
        public void beforeJob(@NonNull JobExecution jobExecution) {

            ElasticSearchRest rest = new ElasticSearchRest(client, mapper);

            // 初始化数据索引
            if (!rest.indexExists(BiDataMartName)) {

                rest.indexCreate(BiDataMartName);
                rest.indexMappingPut(BiDataMartName, "analysis/" + BiDataMartName + ".json", true);
                rest.indexFlush(BiDataMartName, null, null);
            }

            // 初始化最后更新时间索引
            if (!rest.indexExists(BiAnalysisLastName)) {

                rest.indexCreate(BiAnalysisLastName);
                rest.indexMappingPut(BiAnalysisLastName, "analysis/" + BiAnalysisLastName + ".json", true);
                rest.indexFlush(BiAnalysisLastName, null, null);
            }
        }

        @Override
        public void afterJob(@NonNull JobExecution jobExecution) {
            log.debug("last time ---> {}", DateUtils.dateToStr(cache.getLast()));
            if (Objects.nonNull(cache.getCache())) {
                log.debug("cache size -> {}", cache.getCache().size());
                cache.getCache().clear();
            }
            // 回写最后更新时间
            if (jobExecution.getStatus() == BatchStatus.COMPLETED) {
                rest.documentCreate(BiAnalysisLastName, VendorTradingVolume.TableName, AnalysisLastTime.builder().id(VendorTradingVolume.TableName).last(cache.getLast()).build());
                applicationEventPublisher.publishEvent(new VendorTradingVolumeEvent(cache.getLast()));
            }
        }
    };

    @Bean(ReaderName1)
    @StepScope
    public ElasticSearchClzItemReader<AcquireOrganization> reader1() {

        // 获取最后更新时间
        AnalysisLastTime analysisLastTime = rest.documentGet(BiAnalysisLastName, VendorTradingVolume.TableName, AnalysisLastTime.class);
        Date start;
        if (analysisLastTime != null && analysisLastTime.getLast() != null) {
            start = DateUtils.addDays(analysisLastTime.getLast(), 1);
        } else {
            ElasticSearchRest.SearchResultClz<AcquireQuotationGoods> data = rest.searchAsClz(BiWarehouseQuotationGoodsName, ElasticSearchRest.Page.of(1, 1), null, Collections.singleton(SortBuilders.fieldSort("time").order(SortOrder.ASC)), null, AcquireQuotationGoods.class);
            if (Objects.isNull(data) || CollectionUtils.isEmpty(data.getData())) {
                throw new BiException("analysisLastTime init error");
            }
            start = Date.from(data.getData().get(0).getTime().atZone(ZoneId.systemDefault()).toInstant());
        }
        Date[] rangeDay = DateUtils.rangeDay(start);
        cache.setLast(start);
        OrganizationQuery query = OrganizationQuery.builder().architectureType(1).end(rangeDay[1]).del(false).build();
        long count = rest.searchCount(BiWarehouseOrganizationName, BoolQueryBuilders.organization(query));
        log.debug("查询日期 -> {} count -> {}", DateUtils.dateToStr(rangeDay[1], DateUtils.FORMAT_PATTERN_DATE), count);

        //初始化缓存, 不能并行任务
        if (Objects.isNull(cache.getCache())) {
            cache.setCache(new HashMap<>());
        }

        ElasticSearchClzItemReader<AcquireOrganization> reader = new ElasticSearchClzItemReader<>(client, mapper, BiWarehouseOrganizationName,
                MAX_BATCH_NUM, BoolQueryBuilders.organization(query), Collections.singleton(SortBuilders.fieldSort("id").order(SortOrder.ASC)), AcquireOrganization.class);
        reader.setName(ReaderName1);
        return reader;
    }

    @Bean(ProcessName1)
    public ItemProcessor<AcquireOrganization, VendorTradingVolume> processor1() {
        return acquireOrganization -> {
            Date last = cache.getLast();
            LocalDateTime time = LocalDateTime.ofInstant(last.toInstant(), ZoneId.systemDefault());
            String id = EncryptUtils.MD5_32(DateUtils.dateToStr(last, DateUtils.FORMAT_PATTERN_DATE) + acquireOrganization.getId());
            return VendorTradingVolume.builder()
                    .id(id)
                    .organizationName(acquireOrganization.getCompanyName())
                    .organizationId(acquireOrganization.getId())
                    .time(time.toLocalDate())
                    .quotationNum(0.0)
                    .quotationTotal(0.0)
                    .orderGoodsNum(0.0)
                    .orderGoodsTotal(0.0)
                    .refundNum(0.0)
                    .refundTotal(0.0)
                    .build();
        };
    }

    @Bean(WriterName1)
    public ItemWriter<VendorTradingVolume> writer1(){
        return list -> list.forEach(e -> cache.getCache().put(e.getId(), e));
    }

    @Bean(ReaderName2)
    @StepScope
    public ItemReader<VendorTradingVolume> reader2() {
        return EngineUtils.createCacheReader(cache.getCache());
    }

    @Bean(ProcessName2)
    public ItemProcessor<VendorTradingVolume, VendorTradingVolume> processor2() {
        return vendorTradingVolume -> {
            Long organizationId = vendorTradingVolume.getOrganizationId();
            Date[] rangeDay = DateUtils.rangeDay(cache.getLast());
            QuotationGoodsQuery query = QuotationGoodsQuery.builder().statuses(Arrays.asList(3, 4, 5, 6)).start(rangeDay[0]).end(rangeDay[1])
                    .vendorId(organizationId).del(false).build();
            BoolQueryBuilder boolQueryBuilder = BoolQueryBuilders.quotationGoods(query);

            AggregationBuilder idNum = AggregationBuilders.count("id").field("id");
            AggregationBuilder sunTotal = AggregationBuilders.sum("total").field("total");

            SearchResponse search = rest.search(BiWarehouseQuotationGoodsName, ElasticSearchRest.Page.of(1, 0), boolQueryBuilder, null, Arrays.asList(idNum, sunTotal), null);
            Map<String, Aggregation> stringAggregationMap = search.getAggregations().asMap();
            Aggregation id = stringAggregationMap.getOrDefault("id", null);
            Aggregation total = stringAggregationMap.getOrDefault("total", null);
            if (Objects.nonNull(id)) {
                vendorTradingVolume.setQuotationNum((double) ((ValueCount) id).getValue());
            }
            if (Objects.nonNull(total)) {
                vendorTradingVolume.setQuotationTotal(((Sum) total).getValue());
            }
            return vendorTradingVolume;
        };
    }

    @Bean(ProcessName3)
    public ItemProcessor<VendorTradingVolume, VendorTradingVolume> processor3() {
        return vendorTradingVolume -> {
            Long organizationId = vendorTradingVolume.getOrganizationId();
            Date[] rangeDay = DateUtils.rangeDay(cache.getLast());
            VendorOrderQuery vendorOrderQuery = VendorOrderQuery.builder().status(Arrays.asList(1, 2, 3, 5)).vendorId(organizationId).start(rangeDay[0]).end(rangeDay[1]).del(false).build();
            BoolQueryBuilder vendorOrderBuilder = BoolQueryBuilders.vendorOrder(vendorOrderQuery);
            ElasticSearchRest.SearchResultClz<AcquireVendorOrder> searchAsClz = rest.searchAsClz(BiWarehouseVendorOrderName, null, vendorOrderBuilder, null, null, AcquireVendorOrder.class);
            if (Objects.isNull(searchAsClz) || CollectionUtils.isEmpty(searchAsClz.getData())) {
                return vendorTradingVolume;
            }
            List<AcquireVendorOrder> data = searchAsClz.getData();
            List<Long> vendorOrderId = new ArrayList<>();
            for (AcquireVendorOrder order : data) {
                vendorOrderId.add(order.getId());
            }
            BoolQueryBuilder boolQueryBuilder = BoolQueryBuilders.orderGoods(vendorOrderId);

            ValueCountAggregationBuilder num = AggregationBuilders.count("id").field("id");
            SumAggregationBuilder total = AggregationBuilders.sum("total").field("total");

            SearchResponse search = rest.search(BiWarehouseOrderGoodsName, null, boolQueryBuilder, null, Arrays.asList(num, total), null);
            Map<String, Aggregation> stringAggregationMap = search.getAggregations().asMap();
            Aggregation orderGoodsNum = stringAggregationMap.getOrDefault("id", null);
            Aggregation orderGoodsTotal = stringAggregationMap.getOrDefault("total", null);
            if (Objects.nonNull(orderGoodsNum)) {
                vendorTradingVolume.setOrderGoodsNum((double) ((ValueCount) orderGoodsNum).getValue());
            }
            if (Objects.nonNull(orderGoodsTotal)) {
                vendorTradingVolume.setOrderGoodsTotal(((Sum) orderGoodsTotal).getValue());
            }
            return vendorTradingVolume;
        };
    }

    @Bean(ProcessName4)
    public ItemProcessor<VendorTradingVolume, VendorTradingVolume> processor4() {
        return vendorTradingVolume -> {
            Long organizationId = vendorTradingVolume.getOrganizationId();
            Date[] rangeDay = DateUtils.rangeDay(cache.getLast());
            OrderGoodsRefundQuery query = OrderGoodsRefundQuery.builder().vendorId(organizationId).start(rangeDay[0]).end(rangeDay[1]).del(false).build();
            BoolQueryBuilder boolQueryBuilder = BoolQueryBuilders.orderGoodsRefund(query);

            ValueCountAggregationBuilder num = AggregationBuilders.count("id").field("id");
            SumAggregationBuilder total = AggregationBuilders.sum("total").field("total");

            SearchResponse search = rest.search(BiWarehouseOrderGoodsRefundName, null, boolQueryBuilder, null, Arrays.asList(num, total), null);
            Map<String, Aggregation> stringAggregationMap = search.getAggregations().asMap();
            Aggregation refundGoodsNum = stringAggregationMap.getOrDefault("id", null);
            Aggregation refundGoodsTotal = stringAggregationMap.getOrDefault("total", null);
            if (Objects.nonNull(refundGoodsNum)) {
                vendorTradingVolume.setRefundNum((double) ((ValueCount) refundGoodsNum).getValue());
            }
            if (Objects.nonNull(refundGoodsTotal)) {
                vendorTradingVolume.setRefundTotal(((Sum) refundGoodsTotal).getValue());
            }
            return vendorTradingVolume;
        };
    }




    @Bean(WriterName2)
    public ElasticSearchClzItemWriter<VendorTradingVolume> writer2() {

        return new ElasticSearchClzItemWriter<>(client, mapper, BiDataMartName);
    }


    @Bean(JobName)
    public Job job(@Qualifier(StepName1) Step step1, @Qualifier(StepName2) Step step2,@Qualifier(StepName3) Step step3,@Qualifier(StepName4) Step step4) {

        return jobBuilderFactory.get(JobName)
                .incrementer(new RunIdIncrementer())
                .start(step1)
                .next(step2)
                .next(step3)
                .next(step4)
                .listener(jobExecutionListener)
                .build();
    }

    @Bean(StepName1)
    public Step step1() {

        return stepBuilderFactory.get(StepName1)
                .<AcquireOrganization, VendorTradingVolume>chunk(MAX_BATCH_NUM)
                .reader(reader1())
                .processor(processor1())
                .writer(writer1())
                .build();
    }

    @Bean(StepName2)
    public Step step2() {

        return stepBuilderFactory.get(StepName2)
                .<VendorTradingVolume, VendorTradingVolume>chunk(MAX_BATCH_NUM)
                .reader(reader2())
                .processor(processor2())
                .writer(writer1())
                .build();
    }

    @Bean(StepName3)
    public Step step3() {

        return stepBuilderFactory.get(StepName2)
                .<VendorTradingVolume, VendorTradingVolume>chunk(MAX_BATCH_NUM)
                .reader(reader2())
                .processor(processor3())
                .writer(writer1())
                .build();
    }

    @Bean(StepName4)
    public Step step4() {

        return stepBuilderFactory.get(StepName2)
                .<VendorTradingVolume, VendorTradingVolume>chunk(MAX_BATCH_NUM)
                .reader(reader2())
                .processor(processor4())
                .writer(writer2())
                .build();
    }

}
